package com.linkelee;

import com.linkelee.properties.ConsumerBuildProperties;
import com.linkelee.properties.RpcMqConsumerProperties;
import com.linkelee.properties.RpcMqElementKeyBuilder;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消费者构造器
 *
 * @author lsl
 * @since 2019-02-13
 */
@Component
@Slf4j
@Builder
public class MqConsumerBuilder implements InitializingBean {

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private RpcMqConsumerProperties rpcMqConsumerProperties;
    @Autowired
    private ConnectionFactory connectionFactory;
    @Autowired
    private AmqpAdmin amqpAdmin;
    private Map<String, MqConsumerListener> consumerListenerMap = new ConcurrentHashMap<>();

    public void build() {

        String[] beanNames = applicationContext.getBeanNamesForType(MqConsumerListener.class);
        if (beanNames.length == 0) {
            log.info("no consumer need register, consumer build finished.");
        }
        Arrays.asList(beanNames).stream().forEach(beanName -> {
            log.info(beanName + " consumer is starting.");
            try {
                ConsumerBuildProperties buildProperties;
                ChannelAwareMessageListener listener =
                        (ChannelAwareMessageListener) applicationContext.getBean(beanName);
                buildProperties = ConsumerBuildProperties.genProperties(listener, beanName);
                Queue queue = buildQueue(buildProperties);
                Exchange exchange = buildExchange(buildProperties);
                Binding binding = buildRelation(queue, exchange, buildProperties);
                this.check(exchange, queue);
                amqpAdmin.declareQueue(queue);
                amqpAdmin.declareExchange(exchange);
                amqpAdmin.declareBinding(binding);
                SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
                container.setExclusive(rpcMqConsumerProperties.isExposeListenerChannel());
                container.setPrefetchCount(rpcMqConsumerProperties.getPrefetchCount());
                container.setConcurrentConsumers(rpcMqConsumerProperties.getConcurrentConsumers());
                container.setQueueNames(queue.getName());
                container.setAcknowledgeMode(buildProperties.isAutoAck() ?
                        AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL);
                container.setMessageListener(listener);
                container.start();
                log.info(beanName + " consumer has bean started.");
            } catch (Exception e) {
                log.error(beanName + " consumer start defeat.", e);
            }
        });
    }

    private Binding buildRelation(Queue queue, Exchange exchange, ConsumerBuildProperties buildProperties) {
        BindingBuilder.DestinationConfigurer configurer = BindingBuilder.bind(queue);
        if (exchange instanceof DirectExchange) {
            return configurer.to((DirectExchange) exchange).with(buildProperties.getRoutingKey());
        } else if (exchange instanceof FanoutExchange) {
            return configurer.to((FanoutExchange) exchange);
        } else if (exchange instanceof TopicExchange) {
            return configurer.to((TopicExchange) exchange).with(buildProperties.getRoutingKey());
        } else {
            return configurer.to((DirectExchange) exchange).with(buildProperties.getRoutingKey());
        }
    }

    private Queue buildQueue(ConsumerBuildProperties buildProperties) {
        QueueBuilder builder = buildProperties.isDurable() ?
                QueueBuilder.durable(buildProperties.getQueueName()) :
                QueueBuilder.nonDurable(buildProperties.getQueueName());
        if (rpcMqConsumerProperties.isExposeListenerChannel()) {
            builder = builder.exclusive();
        }
        if (buildProperties.isAutoDeleted()) {
            builder = builder.autoDelete();
        }
        return builder.build();
    }

    private Exchange buildExchange(ConsumerBuildProperties buildProperties) {
        ExchangeBuilder builder;
        builder =
                ExchangeTypes.FANOUT.equals(buildProperties.getExchangeType()) ?
                        ExchangeBuilder.fanoutExchange(buildProperties.getExchangeName()) :
                        ExchangeTypes.TOPIC.equals(buildProperties.getExchangeType()) ?
                                ExchangeBuilder.topicExchange(buildProperties.getExchangeName()) :
                                ExchangeTypes.HEADERS.equals(buildProperties.getExchangeType()) ?
                                        ExchangeBuilder.headersExchange(buildProperties.getExchangeType()) :
                                        ExchangeBuilder.directExchange(RpcMqElementKeyBuilder.DEFAULT_EXCHANGE);
        if (buildProperties.isAutoDeleted()) {
            builder = builder.autoDelete();
        }
        builder = builder.durable(buildProperties.isDurable());
        return builder.build();
    }

    public void check(Exchange exchange, Queue queue) {
        if (amqpAdmin == null || connectionFactory == null || exchange == null || queue == null) {
            throw new IllegalArgumentException("rabbitAdmin and connectionFactory should not be null!");
        }
        log.info("choose exchange " +
                (StringUtils.isEmpty(exchange.getName()) ? RpcMqElementKeyBuilder.DEFAULT_EXCHANGE : exchange.getName())
                + " exchange.");
        log.info("choose queue " + queue.getName() + " queue.");
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.build();
    }
}
